package financial.elt;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class MysqlCDCToKafka {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.enableCheckpointing(5000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //感觉在这里写 cdc Table维护比较困难，还是要用yaml的方式去做数据同步
        String mysqlSourceTableDDLConsumerInfo = "CREATE TABLE cdc_source_fin_customer_info\n" +
                "(\n" +
                "    customer_id VARCHAR(36)  ,\n" +
                "    name        VARCHAR(100) ,\n" +
                "    id_card_no  VARCHAR(18)  ,\n" +
                "    phone       VARCHAR(20)  ,\n" +
                "    address     VARCHAR(200) ,\n" +
                "    create_time VARCHAR(19)  ,\n" +
                "    update_time VARCHAR(19)  ,\n" +
                "    dt          VARCHAR(8)   ,\n" +
                "    PRIMARY KEY (customer_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "      'connector' = 'mysql-cdc',\n" +
                "      'hostname' = 'chdp01',\n" +
                "      'port' = '3306',\n" +
                "      'username' = 'root',\n" +
                "      'password' = 'MyPwd123@',\n" +
                "      'database-name' = 'bg2025',\n" +
                "      'table-name' = 'fin_customer_info'\n" +
                "      )";


        String mysqlSinkTableDDLConsumerInfo = "CREATE TABLE cdc_sink_fin_customer_info\n" +
                "(\n" +
                "    customer_id STRING COMMENT '客户唯一ID（主键）',\n" +
                "    name        STRING COMMENT '客户姓名',\n" +
                "    id_card_no  STRING COMMENT '身份证号（唯一约束）',\n" +
                "    phone       STRING COMMENT '联系电话',\n" +
                "    address     STRING COMMENT '联系地址',\n" +
                "    create_time STRING COMMENT '创建时间（格式：yyyy-MM-dd HH:mm:ss）',\n" +
                "    update_time STRING COMMENT '最后更新时间（格式同上）',\n" +
                "    dt STRING,\n" +
                "       PRIMARY KEY (`customer_id`) NOT ENFORCED" +
                ") WITH (\n" +
                "      'connector' = 'upsert-kafka',\n" +
                "      'topic' = 'cdc_sink_fin_customer_info',\n" +
                "      'properties.bootstrap.servers' = 'chdp01:9092',\n" +
                "      'key.format' = 'json',\n" +
                "      'value.format' = 'json'\n" +
                "      )";


        tableEnv.executeSql(mysqlSourceTableDDLConsumerInfo);
        tableEnv.executeSql(mysqlSinkTableDDLConsumerInfo);

        tableEnv.executeSql("insert into cdc_sink_fin_customer_info select * from cdc_source_fin_customer_info").print();
        env.execute("cdc_sink_fin_customer_info");


    }
}
